package com.lagou.es.kafka;

import com.lagou.common.pojo.ProductES;
import com.lagou.common.pojo.Products;
import com.lagou.common.service.ProductService;
import com.lagou.es.repository.ProductRepository;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class KafkaConsumer {

    @Autowired
    private ProductService productService;

    @Autowired
    private ProductRepository repository;



    @KafkaListener(topics = "product",groupId = "group1")
    public void Listener(ConsumerRecord record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC)String topic){
        Optional<Object> value = Optional.ofNullable(record.value());

        if (value.isPresent()){
            //监听Kafka的product主题，收到新消息（新插入商品的id）
            String id = value.get().toString();
            ack.acknowledge();

            //从数据库中查询该id对象的商品对象
            Products products = productService.queryById(Integer.parseInt(id));

            //ES微服务将通过id查询到的商品对象，添加到Elasticsearch的lagou索引库中
            ProductES productES = new ProductES(products.getId(), products.getName(), products.getPrice(), products.getGoodsDesc());
            repository.save(productES);

        }

    }
}
